package com.codejiwei.flink.table;

import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * author: codejiwei
 * date: 2023/8/7
 * desc: stream2table
 **/
public class Flink_Table_API_03 {
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);


        DataStreamSource<String> streamSource = env.readTextFile("src/main/resources/data/ReceiptLog.csv");


        tEnv.createTemporaryView("SourceTable", streamSource);


        tEnv.executeSql("create temporary table SinkTable(f0 STRING) with ('connector' = 'print')");

        tEnv.executeSql("insert into SinkTable select * from SourceTable");



    }
}
